# 导入日志模块，用于记录程序运行时的信息
# Import the logging module to record information during program execution
import logging
# 导入线程模块，用于支持多线程并发执行
# Import the threading module to support multi-threaded concurrent execution
import threading
# 导入时间模块，用于处理时间相关的操作
# Import the time module to handle time-related operations
import time
# 从dataclasses模块导入dataclass，用于创建数据类
# Import dataclass from the dataclasses module to create data classes
from dataclasses import dataclass

# 从myapp.const.mq导入SlowTopicPrefix，表示慢速主题的前缀
# Import SlowTopicPrefix from myapp.const.mq, representing the prefix for slow topics
from myapp.const.mq import SlowTopicPrefix
# 从myapp.third.mq.consumer导入consume，用于消费消息
# Import consume from myapp.third.mq.consumer for consuming messages
from myapp.third.mq.consumer import consume
# 从myapp.utils.exception导入log_exception，用于记录异常日志
# Import log_exception from myapp.utils.exception for logging exceptions
from myapp.utils.exception import log_exception
# 从myapp.utils.time导入format_duration，用于格式化持续时间
# Import format_duration from myapp.utils.time for formatting durations
from myapp.utils.time import format_duration

# 获取一个名为__name__的logger实例
# Get a logger instance named __name__
log = logging.getLogger(__name__)


# 使用dataclass装饰器定义一个消息上下文类
# Define a message context class using the dataclass decorator
@dataclass
class MsgContext:
    # 消息的主题
    # The topic of the message
    topic: str
    # 消息所在的分区
    # The partition where the message resides
    partition: int
    # 消息的偏移量
    # The offset of the message
    offset: int
    # 消息的键
    # The key of the message
    key: str
    # 消息被消费的时间戳
    # The timestamp when the message was consumed
    consume_at: float


# 定义一个消费者类
# Define a Consumer class
class Consumer:
    # 初始化方法
    # Initialization method
    def __init__(self):
        # 用于存储普通消费者的字典
        # Dictionary to store normal consumers
        self.consumers = {}
        # 用于存储慢速主题消费者的字典
        # Dictionary to store slow topic consumers
        self.slow_topic_consumers = {}

    # 定义一个任务装饰器
    # Define a task decorator
    def task(self, *args, **opts):
        # 定义一个装饰器函数
        # Define a decorator function
        def decorator(func):
            # 构建任务信息字典
            # Build a task information dictionary
            task_info = {
                'func': func,
                'args': args,
                'opts': opts or {}
            }
            # 判断是否为慢速主题
            # Determine if it is a slow topic
            slow_topic = True if args[0].startswith(SlowTopicPrefix) else False
            # 根据是否为慢速主题，将任务信息存入相应的字典
            # Store the task information in the corresponding dictionary based on whether it is a slow topic
            if slow_topic:
                self.slow_topic_consumers[args[0]] = task_info
            else:
                self.consumers[args[0]] = task_info
            # 返回原始函数
            # Return the original function
            return func

        # 返回装饰器
        # Return the decorator
        return decorator

    # 定义订阅方法
    # Define the subscribe method
    def subscribe(self, event, slow_topic=False):
        # 设置事件，通知主线程订阅已开始
        # Set the event to notify the main thread that subscription has started
        event.set()
        # 循环消费消息
        # Loop to consume messages
        for message in consume(slow_topic=slow_topic):
            # 记录消息接收时间
            # Record the message reception time
            st = time.time()
            # 解码消息的键
            # Decode the message key
            key = message.key().decode('utf-8')
            # 解码消息的值
            # Decode the message value
            value = message.value().decode('utf-8')
            # 获取消息的主题
            # Get the message topic
            topic = message.topic()
            # 获取消息的分区
            # Get the message partition
            partition = message.partition()
            # 获取消息的偏移量
            # Get the message offset
            offset = message.offset()

            # 记录接收到的消息信息
            # Log the received message information
            log.info(
                f"recv msg, Topic: {topic}, Partition: {partition}, Offset: {offset}, Key: {key}, Value: {value}")

            # 根据是否为慢速主题，从相应的字典中获取任务信息
            # Get task information from the corresponding dictionary based on whether it is a slow topic
            if slow_topic:
                task_info = self.slow_topic_consumers.get(key)
            else:
                task_info = self.consumers.get(key)

            # 如果未找到任务信息，则记录错误并继续
            # If task information is not found, log an error and continue
            if task_info is None:
                log.error(f"未知的 msg_key: {key}")
                continue

            # 获取要执行的函数
            # Get the function to be executed
            func = task_info['func']
            # 使用log_exception上下文管理器来捕获和记录异常
            # Use the log_exception context manager to catch and log exceptions
            with log_exception:
                # 记录任务开始时间
                # Record the task start time
                start_time = time.time()
                # 执行任务函数
                # Execute the task function
                func(MsgContext(topic=topic, partition=partition, offset=offset,
                                key=key,
                                consume_at=st), value)
                # 计算任务执行耗时
                # Calculate the task execution duration
                duration = time.time() - start_time
                # 记录任务耗时信息
                # Log the task duration information
                log.info(
                    f"msg cost:{format_duration(duration)}, Topic: {topic}, Partition: {partition}, Offset: {offset}, Key: {key}")

    # 定义运行方法
    # Define the run method
    def run(self):
        # 创建两个线程事件
        # Create two thread events
        e1 = threading.Event()
        e2 = threading.Eevent()

        # 创建一个普通消费者线程
        # Create a normal consumer thread
        t = threading.Thread(target=self.subscribe, args=(e1,))
        # 创建一个慢速主题消费者线程
        # Create a slow topic consumer thread
        t_slow = threading.Thread(target=self.subscribe, args=(e2,), kwargs={"slow_topic": True})
        # 将线程设置为守护线程
        # Set the threads as daemon threads
        t.daemon = True
        t_slow.daemon = True

        # 启动线程
        # Start the threads
        t.start()
        t_slow.start()

        # 等待事件被设置
        # Wait for the events to be set
        e1.wait()
        e2.wait()

        # 无限循环，检查线程是否存活
        # Infinite loop to check if the threads are alive
        while True:
            # 如果普通消费者线程退出，则记录致命错误
            # If the normal consumer thread exits, log a fatal error
            if t.is_alive() is False:
                log.fatal(f"consumer exited")
            # 如果慢速主题消费者线程退出，则记录错误
            # If the slow topic consumer thread exits, log an error
            if t_slow.is_alive() is False:
                log.error(f"consumer slow_topic exited")

            # 每10秒检查一次
            # Check every 10 seconds
            time.sleep(10)


# 创建一个Consumer实例
# Create a Consumer instance
app = Consumer()